package com.solace.disruptor.producer;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.solace.disruptor.consumer.LongEvent;
import com.solace.disruptor.consumer.LongEventFactory;
import com.solace.disruptor.consumer.LongEventProducer;

import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

@SuppressWarnings("ALL")
public class Test {
    public static void main(String[] args) {
        ThreadFactory producerFactory = Executors.defaultThreadFactory();
        // 创建缓冲池
        LongEventFactory eventFactory = new LongEventFactory();
 
        // 创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
        int ringBufferSize = 1024 * 1024;
        //eventFactory
        Disruptor<LongEvent> disruptor = new Disruptor<>(eventFactory, ringBufferSize, producerFactory,
                ProducerType.SINGLE, new BlockingWaitStrategy());
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
 
        // 创建10个消费者来处理同一个生产者发的消息(这10个消费者不重复消费消息)
        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer();
        }
        disruptor.handleEventsWithWorkerPool(consumers);
 
        disruptor.start();
 
        LongEventProducer longEventProducer = new LongEventProducer(ringBuffer);
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long i = 0; i < 100L; i++) {
            bb.putLong(0, i);
            longEventProducer.onData(bb);
        }
 
        disruptor.shutdown();
    }
}
